package iot.queue;

import iot.DataDispatcher;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.*;

/**
 * Created by dave on 18-7-27 上午11:06.
 */
public class DataQueueDispatcher implements DataDispatcher, InitializingBean {
    @Autowired(required = false) List<DataQueue<?>> queues;
    private final Map<Class<?>, DataQueue<?>> queueMap = new HashMap<>();
    private ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 用于写入数据操作的线程池
    private Timer timer; // 用于定时向各DataQueue发送定时事件消息

    @Override
    public void afterPropertiesSet() throws Exception {
        if (queues != null && queues.size() > 0) {
            queues.forEach(q -> queueMap.put(q.getEntityType(), q));
            int concurrentWrite = queues.stream().map(DataQueue::getConcurrentWrite).reduce(0, (r, i) -> r + i);
            executor.setCorePoolSize((int) (concurrentWrite * 0.5));
            executor.setMaxPoolSize(concurrentWrite);
            executor.setQueueCapacity((int) (concurrentWrite * 1.5));
            executor.afterPropertiesSet();

            timer = new Timer();
            timer.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    queues.forEach(DataQueue::onTimer);
                }
            }, 5000, 100);
        }
    }

    // 根据参数的类型调用对应的队列方法。入队成功返回true；失败返回false，此时应当将数据发送至消息队列
    @Override
    @SuppressWarnings("unchecked")
    public boolean dispatch(Object obj) {
        DataQueue<Object> queue = (DataQueue<Object>) queueMap.get(obj.getClass());
        if (queue != null) {
            return queue.offer(obj);
        }
        return false;
    }

    @SuppressWarnings("unchecked")
    public <T> DataQueue<T> getQueue(Class<T> type) {
        return (DataQueue<T>) queueMap.get(type);
    }

    public ThreadPoolTaskExecutor getExecutor() {
        return executor;
    }
}
